package com.bw.gmall.realtime.dws.app;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.*;

public class RestaurantAnalysis {

    public static void main(String[] args) throws Exception {
        // 设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        // 模拟数据源 - 在实际应用中，这里可以连接到Kafka、MySQL等数据源
        DataStream<String> inputStream = env.fromElements(
                "2025-07-01 10:00:00,user1,杭州西湖店,招牌肉蟹煲,198.0,new,堂食",
                "2025-07-01 10:30:00,user2,上海外滩店,鸡爪煲,158.0,new,外卖",
                "2025-07-01 11:00:00,user1,杭州西湖店,虾兵蟹将煲,188.0,returning,堂食",
                "2025-07-01 12:00:00,user3,北京王府井店,牛蛙煲,168.0,new,外卖",
                "2025-07-01 13:00:00,user4,广州天河店,鸡翅煲,148.0,new,堂食"
        );

        // 解析输入数据
        DataStream<Order> orderStream = inputStream
                .map(new OrderParser())
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner(new SerializableTimestampAssigner<Order>() {
                            @Override
                            public long extractTimestamp(Order order, long recordTimestamp) {
                                return order.getTimestamp();
                            }
                        }));

        // 1. 分析新老用户数量及消费金额
        analyzeUserTypes(orderStream);

        // 2. 分析月度各个店铺的销售额
        analyzeMonthlySalesByStore(orderStream);

        // 3. 分析菜品受欢迎程度
        analyzeDishPopularity(orderStream);

        // 4. 实时检测热门菜品（每分钟更新）
        detectHotDishes(orderStream);

        // 5. 分析不同订单类型（堂食/外卖）的占比
        analyzeOrderTypes(orderStream);

        // 6. 分析用户复购率
        analyzeUserRepurchaseRate(orderStream);

        // 执行作业
        env.execute("胖哥俩肉蟹煲数据分析");
    }

    // 解析订单数据
    public static class OrderParser implements MapFunction<String, Order> {
        @Override
        public Order map(String value) throws Exception {
            String[] fields = value.split(",");
            return new Order(
                    fields[0],
                    fields[1],
                    fields[2],
                    fields[3],
                    Double.parseDouble(fields[4]),
                    fields[5],
                    fields[6]
            );
        }
    }

    // 分析新老用户数量及消费金额
    private static void analyzeUserTypes(DataStream<Order> orderStream) {
        orderStream
                .map(order -> Tuple3.of(order.getUserType(), order.getPrice(), 1L))
                .keyBy(t -> t.f0)
                .timeWindow(Time.days(30))
                .aggregate(new UserTypeAggregateFunction(), new UserTypeWindowFunction())
                .print("新老用户消费分析");
    }

    // 分析月度各个店铺的销售额
    private static void analyzeMonthlySalesByStore(DataStream<Order> orderStream) {
        orderStream
                .map(order -> Tuple3.of(order.getStoreName(), order.getPrice(), 1L))
                .keyBy(t -> t.f0)
                .timeWindow(Time.days(30))
                .aggregate(new SalesAggregateFunction(), new SalesWindowFunction())
                .print("月度店铺销售额统计");
    }

    // 分析菜品受欢迎程度
    private static void analyzeDishPopularity(DataStream<Order> orderStream) {
        orderStream
                .map(order -> Tuple2.of(order.getDishName(), 1L))
                .keyBy(t -> t.f0)
                .timeWindow(Time.days(30))
                .aggregate(new DishPopularityAggregateFunction(), new DishPopularityWindowFunction())
                .print("菜品受欢迎程度统计");
    }

    // 实时检测热门菜品
    private static void detectHotDishes(DataStream<Order> orderStream) {
        orderStream
                .map(order -> Tuple2.of(order.getDishName(), 1L))
                .keyBy(t -> t.f0)
                .timeWindow(Time.minutes(1))
                .process(new TopNDishProcessFunction(3))
                .print("实时热门菜品");
    }

    // 分析不同订单类型（堂食/外卖）的占比
    private static void analyzeOrderTypes(DataStream<Order> orderStream) {
        orderStream
                .map(order -> Tuple2.of(order.getOrderType(), 1L))
                .keyBy(t -> t.f0)
                .timeWindow(Time.days(30))
                .sum(1)
                .print("订单类型占比分析");
    }

    // 分析用户复购率
    private static void analyzeUserRepurchaseRate(DataStream<Order> orderStream) {
        orderStream
                .keyBy(Order::getUserId)
                .flatMap(new RepurchaseRateFunction())
                .keyBy(t -> t.f0)
                .timeWindow(Time.days(30))
                .aggregate(new RepurchaseAggregateFunction(), new RepurchaseWindowFunction())
                .print("用户复购率分析");
    }

    // 用户类型聚合函数
    public static class UserTypeAggregateFunction implements AggregateFunction<Tuple3<String, Double, Long>, Tuple3<Long, Double, Long>, Tuple3<String, Double, Long>> {
        @Override
        public Tuple3<Long, Double, Long> createAccumulator() {
            return Tuple3.of(0L, 0.0, 0L);
        }

        @Override
        public Tuple3<Long, Double, Long> add(Tuple3<String, Double, Long> value, Tuple3<Long, Double, Long> accumulator) {
            return Tuple3.of(accumulator.f0 + 1, accumulator.f1 + value.f1, accumulator.f2 + value.f2);
        }

        @Override
        public Tuple3<String, Double, Long> getResult(Tuple3<Long, Double, Long> accumulator) {
            return Tuple3.of("", accumulator.f1, accumulator.f2);
        }

        @Override
        public Tuple3<Long, Double, Long> merge(Tuple3<Long, Double, Long> a, Tuple3<Long, Double, Long> b) {
            return Tuple3.of(a.f0 + b.f0, a.f1 + b.f1, a.f2 + b.f2);
        }
    }

    // 用户类型窗口函数
    public static class UserTypeWindowFunction implements WindowFunction<Tuple3<String, Double, Long>, Tuple4<String, Double, Long, Double>, String, TimeWindow> {
        @Override
        public void apply(String userType, TimeWindow window, Iterable<Tuple3<String, Double, Long>> input, Collector<Tuple4<String, Double, Long, Double>> out) throws Exception {
            Tuple3<String, Double, Long> agg = input.iterator().next();
            double avgSpend = agg.f1 / agg.f2;
            out.collect(Tuple4.of(userType, agg.f1, agg.f2, avgSpend));
        }
    }

    // 销售额聚合函数 - 修复类型错误
    public static class SalesAggregateFunction implements AggregateFunction<Tuple3<String, Double, Long>, Tuple5<Double, Long, Double, Double, Long>, Tuple5<String, Double, Long, Double, Double>> {
        @Override
        public Tuple5<Double, Long, Double, Double, Long> createAccumulator() {
            return Tuple5.of(0.0, 0L, Double.MIN_VALUE, Double.MAX_VALUE, 0L);
        }

        @Override
        public Tuple5<Double, Long, Double, Double, Long> add(Tuple3<String, Double, Long> value, Tuple5<Double, Long, Double, Double, Long> accumulator) {
            return Tuple5.of(
                    accumulator.f0 + value.f1,
                    accumulator.f1 + value.f2,
                    Math.max(accumulator.f2, value.f1),
                    Math.min(accumulator.f3, value.f1),
                    accumulator.f4 + 1
            );
        }

        @Override
        public Tuple5<String, Double, Long, Double, Double> getResult(Tuple5<Double, Long, Double, Double, Long> accumulator) {
            double avgOrder = accumulator.f4 > 0 ? accumulator.f0 / accumulator.f4 : 0;
            return Tuple5.of("", accumulator.f0, accumulator.f1, accumulator.f2, accumulator.f3);
        }

        @Override
        public Tuple5<Double, Long, Double, Double, Long> merge(Tuple5<Double, Long, Double, Double, Long> a, Tuple5<Double, Long, Double, Double, Long> b) {
            return Tuple5.of(
                    a.f0 + b.f0,
                    a.f1 + b.f1,
                    Math.max(a.f2, b.f2),
                    Math.min(a.f3, b.f3),
                    a.f4 + b.f4
            );
        }
    }

    // 销售额窗口函数 - 修复类型错误和方法调用
// 销售额窗口函数 - 修复类型错误
    public static class SalesWindowFunction implements WindowFunction<Tuple5<String, Double, Long, Double, Double>, Tuple7<String, Double, Long, Double, Double, Double, Long>, String, TimeWindow> {
        @Override
        public void apply(String storeName, TimeWindow window, Iterable<Tuple5<String, Double, Long, Double, Double>> input, Collector<Tuple7<String, Double, Long, Double, Double, Double, Long>> out) throws Exception {
            Tuple5<String, Double, Long, Double, Double> agg = input.iterator().next();
            double growthRate = window.maxTimestamp() > 0 ? (agg.f1 / (window.maxTimestamp() - window.getStart()) * 86400000) : 0;
            long windowSize = (window.maxTimestamp() - window.getStart()) / 1000; // 转换为秒

            // 确保所有字段类型与Tuple7的泛型参数严格匹配
//            out.collect(new Tuple7<>(
//                    storeName,           // T0: String (店铺名称)
//                    agg.f0,              // T1: Double (总销售额)
//                    agg.f1,              // T2: Long (订单数)
//                    agg.f2,              // T3: Double (最大订单金额)
//                    growthRate,          // T4: Double (增长率)
//                    agg.f3,              // T5: Double (最小订单金额)
//                    windowSize           // T6: Long (窗口大小)
//            ));
        }
    }
    // 菜品受欢迎程度聚合函数
    public static class DishPopularityAggregateFunction implements AggregateFunction<Tuple2<String, Long>, Tuple4<Long, Double, Set<String>, Long>, Tuple5<String, Long, Double, Long, Double>> {
        @Override
        public Tuple4<Long, Double, Set<String>, Long> createAccumulator() {
            return Tuple4.of(0L, 0.0, new HashSet<>(), 0L);
        }

        @Override
        public Tuple4<Long, Double, Set<String>, Long> add(Tuple2<String, Long> value, Tuple4<Long, Double, Set<String>, Long> accumulator) {
            accumulator.f2.add(value.f0);
            return Tuple4.of(accumulator.f0 + 1, accumulator.f1 + value.f1, accumulator.f2, accumulator.f3 + value.f1);
        }

        @Override
        public Tuple5<String, Long, Double, Long, Double> getResult(Tuple4<Long, Double, Set<String>, Long> accumulator) {
            double avgOrder = accumulator.f0 > 0 ? accumulator.f3 / accumulator.f0 : 0;
            return Tuple5.of("", accumulator.f0, accumulator.f1, (long) accumulator.f2.size(), avgOrder);
        }

        @Override
        public Tuple4<Long, Double, Set<String>, Long> merge(Tuple4<Long, Double, Set<String>, Long> a, Tuple4<Long, Double, Set<String>, Long> b) {
            a.f2.addAll(b.f2);
            return Tuple4.of(a.f0 + b.f0, a.f1 + b.f1, a.f2, a.f3 + b.f3);
        }
    }

    // 菜品受欢迎程度窗口函数 - 修复方法调用
    public static class DishPopularityWindowFunction implements WindowFunction<Tuple5<String, Long, Double, Long, Double>, Tuple6<String, Long, Double, Long, Double, Double>, String, TimeWindow> {
        @Override
        public void apply(String dishName, TimeWindow window, Iterable<Tuple5<String, Long, Double, Long, Double>> input, Collector<Tuple6<String, Long, Double, Long, Double, Double>> out) throws Exception {
            Tuple5<String, Long, Double, Long, Double> agg = input.iterator().next();
            double orderPerDay = (double) agg.f1 / 30;
            double windowSize = (window.maxTimestamp() - window.getStart()) / (1000 * 3600 * 24); // 转换为天
            out.collect(new Tuple6<>(
                    dishName,
                    agg.f1,
                    agg.f2,
                    agg.f3,
                    orderPerDay,
                    windowSize
            ));
        }
    }

    // 热门菜品TopN处理函数
    public static class TopNDishProcessFunction extends ProcessWindowFunction<Tuple2<String, Long>, String, String, TimeWindow> {
        private final int topSize;

        public TopNDishProcessFunction(int topSize) {
            this.topSize = topSize;
        }

        @Override
        public void process(String key, Context context, Iterable<Tuple2<String, Long>> elements, Collector<String> out) throws Exception {
            List<Tuple2<String, Long>> dishes = new ArrayList<>();
            elements.forEach(dishes::add);

            // 按销量排序
            dishes.sort((d1, d2) -> Long.compare(d2.f1, d1.f1));

            // 构建热门菜品排名
            StringBuilder result = new StringBuilder();
            result.append("===== 热门菜品排名 =====").append("\n");
            for (int i = 0; i < Math.min(topSize, dishes.size()); i++) {
                Tuple2<String, Long> dish = dishes.get(i);
                result.append("No.").append(i + 1).append(": ")
                        .append(dish.f0).append(" - ")
                        .append(dish.f1).append("单")
                        .append("\n");
            }
            out.collect(result.toString());
        }
    }

    // 用户复购率函数
    public static class RepurchaseRateFunction extends RichFlatMapFunction<Order, Tuple2<String, Integer>> {
        private transient ValueState<Integer> orderCountState;

        @Override
        public void open(Configuration parameters) throws Exception {
            orderCountState = getRuntimeContext().getState(
                    new ValueStateDescriptor<>("orderCount", Integer.class, 0)
            );
        }

        @Override
        public void flatMap(Order order, Collector<Tuple2<String, Integer>> out) throws Exception {
            int count = orderCountState.value();
            count++;
            orderCountState.update(count);

            // 如果订单数大于1，表示有复购
            if (count > 1) {
                out.collect(Tuple2.of("repurchase", 1));
            }
            out.collect(Tuple2.of("total", 1));
        }
    }

    // 复购率聚合函数
    public static class RepurchaseAggregateFunction implements AggregateFunction<Tuple2<String, Integer>, Tuple2<Integer, Integer>, Tuple3<String, Double, Integer>> {
        @Override
        public Tuple2<Integer, Integer> createAccumulator() {
            return Tuple2.of(0, 0);
        }

        @Override
        public Tuple2<Integer, Integer> add(Tuple2<String, Integer> value, Tuple2<Integer, Integer> accumulator) {
            if ("repurchase".equals(value.f0)) {
                return Tuple2.of(accumulator.f0 + 1, accumulator.f1);
            } else {
                return Tuple2.of(accumulator.f0, accumulator.f1 + 1);
            }
        }

        @Override
        public Tuple3<String, Double, Integer> getResult(Tuple2<Integer, Integer> accumulator) {
            double rate = accumulator.f1 > 0 ? (double) accumulator.f0 / accumulator.f1 : 0;
            return Tuple3.of("repurchase_rate", rate, accumulator.f1);
        }

        @Override
        public Tuple2<Integer, Integer> merge(Tuple2<Integer, Integer> a, Tuple2<Integer, Integer> b) {
            return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
        }
    }

    // 复购率窗口函数
    public static class RepurchaseWindowFunction implements WindowFunction<Tuple3<String, Double, Integer>, Tuple4<String, Double, Integer, String>, String, TimeWindow> {
        @Override
        public void apply(String key, TimeWindow window, Iterable<Tuple3<String, Double, Integer>> input, Collector<Tuple4<String, Double, Integer, String>> out) throws Exception {
            Tuple3<String, Double, Integer> agg = input.iterator().next();
            String level = agg.f1 >= 0.5 ? "高" : (agg.f1 >= 0.3 ? "中" : "低");
            out.collect(new Tuple4<>(
                    agg.f0,
                    agg.f1,
                    agg.f2,
                    level
            ));
        }
    }

    // 订单类
    public static class Order {
        private String timestamp;
        private String userId;
        private String storeName;
        private String dishName;
        private double price;
        private String userType; // new or returning
        private String orderType; // 堂食 or 外卖

        public Order() {}

        public Order(String timestamp, String userId, String storeName, String dishName, double price, String userType, String orderType) {
            this.timestamp = timestamp;
            this.userId = userId;
            this.storeName = storeName;
            this.dishName = dishName;
            this.price = price;
            this.userType = userType;
            this.orderType = orderType;
        }

        public long getTimestamp() {
            try {
                // 简单解析时间戳，实际应用中可能需要更复杂的解析
                return java.time.LocalDateTime
                        .parse(timestamp)
                        .atZone(java.time.ZoneId.systemDefault())
                        .toInstant()
                        .toEpochMilli();
            } catch (Exception e) {
                return System.currentTimeMillis();
            }
        }

        public String getUserId() {
            return userId;
        }

        public String getStoreName() {
            return storeName;
        }

        public String getDishName() {
            return dishName;
        }

        public double getPrice() {
            return price;
        }

        public String getUserType() {
            return userType;
        }

        public String getOrderType() {
            return orderType;
        }

        @Override
        public String toString() {
            return "Order{" +
                    "timestamp='" + timestamp + '\'' +
                    ", userId='" + userId + '\'' +
                    ", storeName='" + storeName + '\'' +
                    ", dishName='" + dishName + '\'' +
                    ", price=" + price +
                    ", userType='" + userType + '\'' +
                    ", orderType='" + orderType + '\'' +
                    '}';
        }
    }

    // 辅助元组类 - 用于解决类型冲突
    public static class Tuple7<T0, T1, T2, T3, T4, T5, T6> {
        public T0 f0;
        public T1 f1;
        public T2 f2;
        public T3 f3;
        public T4 f4;
        public T5 f5;
        public T6 f6;

        public Tuple7(T0 f0, T1 f1, T2 f2, T3 f3, T4 f4, T5 f5, T6 f6) {
            this.f0 = f0;
            this.f1 = f1;
            this.f2 = f2;
            this.f3 = f3;
            this.f4 = f4;
            this.f5 = f5;
            this.f6 = f6;
        }
    }
}